package org.atlantis.evnlib.task;

import java.util.Collection;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.atlantis.evnlib.exception.EventException;
import org.atlantis.evnlib.execution.RenamingRunnable;
import org.atlantis.evnlib.execution.rejecter.RejectedTaskHandler;

/**
 * 任务池。
 * 
 * @author <a href="mailto:seenoevil.cn@gmail.com">李曦</a><br/>
 *         Created at: 2011-6-30 下午09:28:37
 */
class TaskPool {

	private final RejectedTaskHandler rejectedTaskHandler;
	private final BlockingQueue<Runnable> taskQueue;
	private final int fixedQuotas;
	private final int flexQuotas;
	private final String name;
	private final ExecutorService realExec;
	private final AtomicBoolean closed = new AtomicBoolean();

	private final AtomicInteger fixedWorkingCounter = new AtomicInteger();
	private final AtomicInteger flexWorkingCounter = new AtomicInteger();

	TaskPool(ExecutorService realExec, String name, int fixedQuotas, int flexQuotas, RejectedTaskHandler rejectedTaskHandler,
			int taskQueueSize) {
		this.rejectedTaskHandler = rejectedTaskHandler;
		this.taskQueue = new LinkedBlockingQueue<Runnable>(taskQueueSize);
		this.fixedQuotas = fixedQuotas;
		this.flexQuotas = flexQuotas;
		this.name = name;
		this.realExec = realExec;
		startFixedThread();
	}

	boolean submit(Runnable task) {
		if (closed.get()) throw new EventException("TaskPool<%s> closed.", name);
		boolean submitted = taskQueue.offer(task);
		if (!submitted) submitted = rejectedTaskHandler.rejectedTask(name, task, taskQueue);
		if (submitted) startFlexThread();
		return submitted;
	}

	Collection<Runnable> close() {
		closed.set(true);
		return taskQueue;
	}

	private void startFixedThread() {
		for (int i = 0; i < fixedQuotas; i++)
			realExec.execute(new RenamingRunnable(new FixedQuotasRunner(), String.format("evnlib:%s(fixed)", name)));
	}

	private void startFlexThread() {
		if (flexQuotas == 0) return;
		int taskNum = taskQueue.size();
		int fixedRemainSize = fixedQuotas - fixedWorkingCounter.get();
		if (taskNum <= fixedRemainSize) return;
		if (flexWorkingCounter.incrementAndGet() <= flexQuotas) realExec.execute(new RenamingRunnable(new FlexQuotasRunner(),
				String.format("evnlib:%s(flex)", name)));
		else flexWorkingCounter.decrementAndGet();
	}

	private final class FixedQuotasRunner implements Runnable {
		@Override
		public void run() {
			while (!closed.get()) {
				try {
					Runnable task = taskQueue.take();
					fixedWorkingCounter.incrementAndGet();
					task.run();
					fixedWorkingCounter.decrementAndGet();
					startFlexThread();
				} catch (Throwable cause) {
					TaskExceptionHandler.HandlerRegister.handler.handleException(cause);
				}
			}
		}
	}

	private final class FlexQuotasRunner implements Runnable {
		@Override
		public void run() {
			try {
				if (closed.get()) return;
				Runnable task = taskQueue.poll();
				if (task != null) task.run();
				flexWorkingCounter.decrementAndGet();
				startFlexThread();
			} catch (Throwable cause) {
				TaskExceptionHandler.HandlerRegister.handler.handleException(cause);
			}
		}
	}
}